#**** BEGIN LICENSE BLOCK *****
# Version: MPL 1.1/GPL 2.0/LGPL 2.1
# 
# The contents of this file are subject to the Mozilla Public License
# Version 1.1 (the "License"); you may not use this file except in
# compliance with the License. You may obtain a copy of the License at
# http://www.mozilla.org/MPL/
# 
# Software distributed under the License is distributed on an "AS IS"
# basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
# License for the specific language governing rights and limitations
# under the License.
# 
# The Original Code is pyxpcomext code.
# 
# The Initial Developer of the Original Code is Todd Whiteman.
# Portions created by the Initial Developer are Copyright (C) 2007-2008.
# All Rights Reserved.
# 
# Contributor(s):
#   Todd Whiteman <twhitema@gmail.com> (original author)
# 
# Alternatively, the contents of this file may be used under the terms of
# either the GNU General Public License Version 2 or later (the "GPL"), or
# the GNU Lesser General Public License Version 2.1 or later (the "LGPL"),
# in which case the provisions of the GPL or the LGPL are applicable instead
# of those above. If you wish to allow use of your version of this file only
# under the terms of either the GPL or the LGPL, and not to allow others to
# use your version of this file under the terms of the MPL, indicate your
# decision by deleting the provisions above and replace them with the notice
# and other provisions required by the GPL or the LGPL. If you do not delete
# the provisions above, a recipient may use your version of this file under
# the terms of any one of the MPL, the GPL or the LGPL.
# 
#**** END LICENSE BLOCK *****

#
# An implementation of secure ftp (SFTP). To be used in conjuction with the
# Mozilla protocol service.
#
# This uses "paramiko", a pure python implementation of the 
# SSH2 protocol, of which, we are using the SFTP component.
# http://www.lag.net/paramiko/
#

import os
from os.path import join, exists
import sys
import stat
import socket
import threading
import logging
import time
from time import strftime, gmtime
from urllib import quote

from xpcom import components, ServerException, nsError
from xpcom._xpcom import getProxyForObject, NS_PROXY_SYNC, NS_PROXY_ALWAYS


# Ensure the pylib directories are on the path, required for the
# paramiko and Crypto imports.
extDir = os.path.dirname(__file__)
# Work out the possible platform specific directories.
xulRuntimeSvc = components.classes["@mozilla.org/xre/app-info;1"].\
                    getService(components.interfaces.nsIXULRuntime)
xpcom_os = xulRuntimeSvc.OS
xpcom_os_and_abi = "%s_%s" % (xpcom_os, xulRuntimeSvc.XPCOMABI)
# Add the paths.
possiblePylibPaths = [join(extDir, "pylib"),
                      join(extDir, "platform", xpcom_os, "pylib"),
                      join(extDir, "platform", xpcom_os_and_abi, "pylib")]
for pylibPath in possiblePylibPaths:
    if exists(pylibPath) and pylibPath not in sys.path:
        print "Adding pylib to sys.path: %r" % (pylibPath, )
        sys.path.append(pylibPath)


import paramiko
import paramiko.sftp
from paramiko.ssh_exception import SSHException, AuthenticationException, \
                                   BadAuthenticationType, PartialAuthentication

# Basic logging to stdout. TODO: Use the nsIConsole logging service?
logging.basicConfig(stream=sys.stdout)
log = logging.getLogger("SFTP Protocol Handler")
#log.setLevel(logging.DEBUG)


###########################################################
###                 SSH Input Streams                   ###
###########################################################


class SSHInputStream(object):
    _com_interfaces_ = [components.interfaces.nsIInputStream]

    def __init__(self, initial_data=None):
        self.data = initial_data or ''
        self._non_blocking = False
        self._lock = threading.Lock()

    def close( self ):
        log.debug("SSHInputStream:: close")
        pass

    def isNonBlocking(self):
        log.debug("SSHInputStream:: isNonBlocking")
        return self._non_blocking

    def available( self ):
        #log.debug("SSHInputStream:: available")
        return len(self.data)

    def read( self, amount):
        #log.debug("SSHInputStream:: read amount: %d", amount)
        self._lock.acquire()
        try:
            data = self.data[:amount]
            self.data = self.data[amount:]
            return data
        finally:
            self._lock.release()
            pass

    readStr=read

    # Data is stored in encoded UTF-8 format.
    def write( self, data ):
        #log.debug("SSHInputStream:: write")
        self._lock.acquire()
        try:
            self.data += data.encode("utf-8")
        finally:
            self._lock.release()

    writeStr=write

    def get_observer( self ):
        log.debug("SSHInputStream:: get_observer")
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)

    def set_observer( self, param0 ):
        log.debug("SSHInputStream:: set_observer")
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)

class SSHBinaryInputStream(SSHInputStream):
    _com_interfaces_ = [components.interfaces.nsIInputStream,
                        components.interfaces.nsIBinaryInputStream]

    def write( self, data ):
        #log.debug("SSHBinaryInputStream:: write")
        self._lock.acquire()
        try:
            self.data += data
        finally:
            self._lock.release()

    ## nsIBinaryInputStream methods

    def setInputStream(self, aInputStream):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)

    ##
    # Read 8-bits from the stream.
    #
    # @return that byte to be treated as a boolean.
    def readBoolean(self):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
    def read8(self):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
    def read16(self):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
    def read32(self):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
    def read64(self):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
    def readFloat(self):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
    def readDouble(self):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
 
    ##
    # Read an 8-bit pascal style string from the stream.
    # 32-bit length field, followed by length 8-bit chars.
    def readCString(self):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
 
    ##
    # Read an 16-bit pascal style string from the stream.
    # 32-bit length field, followed by length PRUnichars.
    def readString(self):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
 
    ##
    # Read an opaque byte array from the stream.
    #
    # @param aLength the number of bytes that must be read.
    #
    # @throws NS_ERROR_FAILURE if it can't read aLength bytes
    def readBytes(self, aLength):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)
 
    ##
    # Read an opaque byte array from the stream, storing the results
    # as an array of PRUint8s.
    #
    # @param aLength the number of bytes that must be read.
    #
    # @throws NS_ERROR_FAILURE if it can't read aLength bytes
    def readByteArray(self, aLength):
        raise ServerException(nsError.NS_ERROR_NOT_IMPLEMENTED)


# These two binary input streams are no different to the binary inout stream,
# except they can be later QueryInterfaced to check if the data stream is a
# stdout or a stderr stream.
class SSHStdoutStream(SSHBinaryInputStream):
    _com_interfaces_ = [components.interfaces.nsIInputStream,
                        components.interfaces.nsIBinaryInputStream,
                        components.interfaces.pyISSHStdoutInputStream]

class SSHStderrStream(SSHBinaryInputStream):
    _com_interfaces_ = [components.interfaces.nsIInputStream,
                        components.interfaces.nsIBinaryInputStream,
                        components.interfaces.pyISSHStderrInputStream]


###########################################################
###               AuthInfo for Login                    ###
###########################################################


# Since nobody implements nsIAuthInformation, provide our own class.
class sftpAuthInfo(object):
    _com_interfaces_ = [components.interfaces.nsIAuthInformation]

    flags = components.interfaces.nsIAuthInformation.AUTH_HOST

    def __init__(self, username="", password="", flags=None, realm="",
                 domain="", authenticationScheme=""):
        self.username = username
        self.password = password
        if flags is not None:
            self.flags = flags
        self.realm = realm
        self.domain = domain
        self.authenticationScheme = authenticationScheme


###########################################################
###               Connection caching                    ###
###########################################################


# Global timer object, as I could not get this to work successfully as
# properties on the sftpConnectionCache class.
timer_running = False
timer = components.classes["@mozilla.org/timer;1"].\
                createInstance(components.interfaces.nsITimer)

# Connection cache, a singleton instance to cache sftp connections. This
# cache will periodically remove/close connections after they have been
# dormant for a period of time.
class sftpConnectionCache(object):
    _com_interfaces_ = [components.interfaces.nsITimerCallback]

    # TODO: Pref this expiry value, or get from existing mozilla prefs?
    EXPIRY_PERIOD = 15 * 60  # Fifteen minutes.

    def __init__(self):
        self._cache = {}
        self._lock = threading.Lock()

    def notify(self, aTimer=None):
        # Periodically, clear old connections from the cache.
        log.debug("cache:: Got timer notify event")
        self._lock.acquire()
        try:
            for cache_key, cache_item in self._cache.items():
                transport, user, cache_time = cache_item
                if (time.time() - cache_time) > self.EXPIRY_PERIOD:
                    log.debug("cache:: expired the cached connection for: %r", cache_key)
                    self._cache.pop(cache_key, None)
            if len(self._cache) == 0:
                # Nothing in the cache, no point in keeping the timer active.
                global timer
                global timer_running
                timer.cancel()
                timer_running = False
                log.debug("cache:: cancelled the timer, nothing left in the cache")
        finally:
            self._lock.release()

    def _ensureCleanupTimerActivated(self):
        global timer
        global timer_running
        if not timer_running:
            # Fire the notify method every 1 minute.
            timer.initWithCallback(self, 1 * 60 * 1000, components.interfaces.nsITimer.TYPE_REPEATING_SLACK)
            timer_running = True
            log.debug("cache:: created a cache timer")

    ##
    # Add a connection to cache.
    # @param transport {paramiko.Transport}  The sftp connection to cache.
    # @param host {str}  The host name for the connection.
    # @param port {int}  The port number to use for the connection.
    # @param username {str}  The username used to login with.
    #
    def addConnection(self, transport, host, port, username):
        self._lock.acquire()
        try:
            cache_key = "%s:%s" % (host, port)
            self._cache[cache_key] = (transport, username, time.time())
            log.debug("cache:: cached connection for: %r", cache_key)
            self._ensureCleanupTimerActivated()
        finally:
            self._lock.release()

    ##
    # Add a connection to cache.
    # @param host {str}  The host name for the connection.
    # @param port {int}  The port number to use for the connection.
    # @param username {str}  The username used to login with.
    # @returns {paramiko.Transport}  The sftp connection.
    #
    def getConnection(self, host, port, username):
        self._lock.acquire()
        try:
            cache_key = "%s:%s" % (host, port)
            cache_item = self._cache.get(cache_key)
            if cache_item:
                # There is a cached item, check if it's still alive.
                transport, user, cache_time = cache_item
                if transport.is_active():
                    if username and username != user:
                        # Is setup for a different username.
                        return None
                    log.debug("cache:: found cached connection for: %r", cache_key)
                    # Update the last cached time.
                    self._cache[cache_key] = (transport, user, time.time())
                    return transport
                # Else, it was closed, remove this cached entry.
                self._cache.pop(cache_key, None)
            return None
        finally:
            self._lock.release()


# Create a global instance of the cache service.
connectionCache =  sftpConnectionCache()


###########################################################
###                 SFTP Connection                     ###
###########################################################


#
# There are two transport connections established for an sftp connection. One is
# used for the command channel, and the other for the data channel. The command
# channel is the first connection made and is used to negotiate the second,
# data, channel. The data channel is driven by the command channel and is
# initiated by the server. Client initiation is the most common case and is
# attempted first.
#
class pythonextSSHConnection(threading.Thread):
    _com_interfaces_ = [components.interfaces.pyISSHConnection,
                        components.interfaces.pyISSHCommand,
                        components.interfaces.pyISSHInteraction]
    _reg_contractid_ = '@pythonext.mozdev.org/sshConnection;1'
    _reg_clsid_ = '{d036ecde-f0e1-41e5-8b41-dcfae99151b4}'
    _reg_desc_ = "Paramiko SSH connection"

        # From nsMimeTypes.h
    UNKNOWN_CONTENT_TYPE = "application/x-unknown-content-type"
    APPLICATION_GUESS_FROM_EXT = "application/x-vnd.mozilla.guess-from-ext"
    APPLICATION_HTTP_INDEX_FORMAT = "application/http-index-format"

        # Error code not available in nsError.
    NS_ERROR_NOT_RESUMABLE = nsError.NS_ERROR_GENERATE_FAILURE(nsError.NS_ERROR_MODULE_NETWORK, 25)

        # How much data to read per block.
    READ_BLOCK_SIZE = 4096

        # The type of async operation being performed.
    TYPE_ASYNC_OPEN = 1
    TYPE_ASYNC_COMMAND = 2
    TYPE_ASYNC_INTERACT = 3

        # Different status modes the channel may be in.
    STATE_ERROR = -1
    STATE_INITIALIZED = 0
    STATE_OPENING = 1
    STATE_PROCESSING = 2
    STATE_SUSPENDED = 3
    STATE_CANCELLED = 4
    STATE_CLOSED = 5

    # Internally used attributes.

        # The type of async operation, one of TYPE_ASYNC_* above.
    _async_type = None
        # Listener gets set in the asyncOpen() call.
    _listener = None
        # Context gets set in the asyncOpen() call.
    _context = None
        # The remote command to run, used by asyncRunCommand().
    _command = None
        # The terminal type, used by asyncInteract().
    _termType = None
        # The thread who initially calls asyncOpen(). All notification events
        # will be sent back using this thread's event queue.
    _targetThread = None
        # A notification event for state changes.
    _state_notification = None
        # The sftp connection object.
    _sftp = None
        # The sftp file handle.
    _sftp_file = None
        # The channel used for running interactively.
    _channel = None
        # Where to start download the file from.
    _startPos = 0
        # Whether the onStartRequest notification has been sent.
    _sentStartRequest = False
        # Whether this request is for a directory listing.
    isDirectoryListing = False
        # Length of the file being uploaded/download.
    fileSize = 0
        # File information, to be used with resumeAt.
    entityID = None
        # The status associated with the request.
    status = nsError.NS_OK
        # The URI corresponding to the channel.  Its value is immutable.
    URI = None
        # Content type of the data being sent/received.    
    contentType = APPLICATION_GUESS_FROM_EXT
        # Character encoding used by the data.
    contentCharset = ""
        # Length of data to upload/download.
    contentLength = -1
        # Stream used for data communication through the listener.
    contentStream = None
        # Upload stream used for uploading data.
    uploadStream = None

        # Static class variable used for logging purposes.
    _session_id = 1

        ##
        # Create a new sftp channel from the given uri.
        # @param uri {components.interfaces.nsIURI}
        #
    def __init__(self):
        # Create a unique id, only used for logging purposes.
        self._id = pythonextSSHConnection._session_id
        pythonextSSHConnection._session_id += 1

        # Initialize the thread.
        threading.Thread.__init__(self, name="Paramiko SFTP connection handler")
        # Ensure the main thread does not wait on us to finish before quitting.
        self.setDaemon(True)

        # Set the initial state.
        self._state_notification = threading.Condition()
        self.__internal_state = self.STATE_INITIALIZED

    def _get_state(self):
        return self.__internal_state
    def _set_state(self, val):
        self._state_notification.acquire()
        try:
            if val != self.__internal_state:
                self.__internal_state = val
                self._state_notification.notifyAll()
        finally:
            self._state_notification.release()
        ##
        # The internal status of what the sftp handler is doing.
        # 
    _state = property(_get_state, _set_state)

    logging_DEBUG = logging.DEBUG
    def debug(self, message, *args, **kwargs):
        if log.level == self.logging_DEBUG:
            log.debug("session: %d: %s" % (self._id, message), *args, **kwargs)

        ##
        # Sets the URI for this channel.
        #
    def setURI(self, aURI):
        self.debug("setURI")
        if aURI is not None:
            aURI = aURI.clone()
        self.URI = aURI

        ##
        # Asynchronously open this channel.  Data is fed to the specified
        # listener as it becomes available.  The listener methods are
        # called on the thread that calls asyncOpen and are not called until
        # after asyncOpen returns.  If asyncOpen returns successfully, the
        # channel promises to call at least onStartRequest and onStopRequest.
        #
        # If asyncOpen returns successfully, the channel is responsible for
        # keeping itself alive until it has called onStopRequest on aListener or
        # called onChannelRedirect.
        #
        # NOTE: Implementations should throw NS_ERROR_ALREADY_OPENED if the
        # channel is reopened.
        #
        # @param aListener the nsIStreamListener implementation
        # @param aContext an opaque parameter forwarded to aListener's methods
        #
        # nsIChannel : void asyncOpen(in nsIStreamListener aListener, in nsISupports aContext);
    def asyncOpen(self, aListener, aContext):
        self.debug("asyncOpen")
        if self._state != self.STATE_INITIALIZED:
            raise ServerException(nsError.NS_ERROR_ALREADY_OPENED)

        self._async_type = self.TYPE_ASYNC_OPEN
        self._listener = getProxyForObject(self._targetThread,
                                           components.interfaces.pyISSHStreamListener,
                                           aListener,
                                           NS_PROXY_SYNC | NS_PROXY_ALWAYS)
        self._context = aContext
        self._state = self.STATE_OPENING

        threadMgr = components.classes["@mozilla.org/thread-manager;1"].\
                        getService(components.interfaces.nsIThreadManager)
        self._targetThread = threadMgr.currentThread

        # Kick off the asynchronous Python thread (this calls the run() method).
        self.start()

    ##################################
    #   nsIRequest methods
    ##################################

        ##
        # Indicates whether the request is pending. nsIRequest::isPending is
        # true when there is an outstanding asynchronous event that will make
        # the request no longer be pending.  Requests do not necessarily start
        # out pending; in some cases, requests have to be explicitly initiated
        # (e.g. nsIChannel implementations are only pending once asyncOpen
        # returns successfully).
        #
        # Requests can become pending multiple times during their lifetime.
        #
        # @return TRUE if the request has yet to reach completion.
        # @return FALSE if the request has reached completion (e.g., after
        #   OnStopRequest has fired).
        # @note Suspended requests are still considered pending.
    def isPending(self):
        self.debug("isPending")
        return self._state in (self.STATE_OPENING, self.STATE_PROCESSING,
                               self.STATE_SUSPENDED)

        ##
        # Cancels the current request.  This will close any open input or
        # output streams and terminate any async requests.  Users should 
        # normally pass NS_BINDING_ABORTED, although other errors may also
        # be passed.  The error passed in will become the value of the 
        # status attribute.
        #
        # Implementations must not send any notifications (e.g. via
        # nsIRequestObserver) synchronously from this function. Similarly,
        # removal from the load group (if any) must also happen asynchronously.
        #
        # Requests that use nsIStreamListener must not call onDataAvailable
        # anymore after cancel has been called.
        #
        # @param aStatus the reason for canceling this request.
        #
        # NOTE: most nsIRequest implementations expect aStatus to be a
        # failure code; however, some implementations may allow aStatus to
        # be a success code such as NS_OK.  In general, aStatus should be
        # a failure code.
        #
        # nsIRequest : void cancel(in nsresult aStatus);
    def cancel(self, aStatus):
        self.debug("cancel")
        self.status = aStatus
        self._state = self.STATE_CANCELLED
        if self._channel:
            if self._async_type == self.TYPE_ASYNC_INTERACT:
                self._channel.close()

        ##
        # Suspends the current request.  This may have the effect of closing
        # any underlying transport (in order to free up resources), although
        # any open streams remain logically opened and will continue delivering
        # data when the transport is resumed.
        #
        # Calling cancel() on a suspended request must not send any
        # notifications (such as onstopRequest) until the request is resumed.
        #
        # NOTE: some implementations are unable to immediately suspend, and
        # may continue to deliver events already posted to an event queue. In
        # general, callers should be capable of handling events even after 
        # suspending a request.
        #
        # nsIRequest : void suspend();
    def suspend(self):
        self.debug("suspend")
        self._state = self.STATE_SUSPENDED

        ##
        # Resumes the current request.  This may have the effect of re-opening
        # any underlying transport and will resume the delivery of data to 
        # any open streams.
        #
        # nsIRequest : void resume();
    def resume(self):
        self.debug("resume")
        self._state = self.STATE_PROCESSING

    ##################################
    #   nsIResumableChannel methods
    ##################################

        ##
        # Prepare this channel for resuming. The request will not start until
        # asyncOpen or open is called. Calling resumeAt after open or asyncOpen
        # has been called has undefined behaviour.
        #
        # @param startPos the starting offset, in bytes, to use to download
        # @param entityID information about the file, to match before obtaining
        #  the file. Pass an empty string to use anything.
        #
        # During OnStartRequest, this channel will have a status of
        #  NS_ERROR_NOT_RESUMABLE if the file cannot be resumed, eg because the
        #  server doesn't support this. This error may occur even if startPos
        #  is 0, so that the front end can warn the user.
        # Similarly, the status of this channel during OnStartRequest may be
        #  NS_ERROR_ENTITY_CHANGED, which indicates that the entity has changed,
        #  as indicated by a changed entityID.
        # In both of these cases, no OnDataAvailable will be called, and
        #  OnStopRequest will immediately follow with the same status code.
    def resumeAt(self, aStartPos, aEntityID):
        self._startPos = aStartPos
        self.entityID = aEntityID

    ##################################
    #   nsIUploadChannel methods
    ##################################

        ##
        # Sets a stream to be uploaded by this channel.
        # 
        # @param aStream
        #        The stream to be uploaded by this channel.
        # @param aContentType
        #        If aContentType is empty, the protocol will assume that no
        #        content headers are to be added to the uploaded stream and that
        #        any required headers are already encoded in the stream.  In the
        #        case of HTTP, if this parameter is non-empty, then its value will
        #        replace any existing Content-Type header on the HTTP request.
        #        In the case of FTP and FILE, this parameter is ignored.
        # @param aContentLength
        #        A value of -1 indicates that the length of the stream should be
        #        determined by calling the stream's |available| method.
        #
        # nsIUploadChannel : void setUploadStream(in nsIInputStream aStream,
        #                                         in ACString aContentType,
        #                                         in long aContentLength);
    def setUploadStream(self, aStream, aContentType, aContentLength):
        self.debug("setUploadStream")
        self.uploadStream = aStream
        self.contentType = aContentType
        self.contentLength = aContentLength

    ##################################
    #   internal helper functions
    ##################################

    def _startRequest(self, contentType=None, actionType=None):
        if contentType:
            self.contentType = contentType

        if actionType == "run_command":
            self.contentStream = SSHStdoutStream()
        elif self.isDirectoryListing:
            self.contentStream = SSHInputStream()
        else:
            self.contentStream = SSHBinaryInputStream()

        self._sentStartRequest = True
        self._listener.onStartRequest(None, self._context)

    def _dataAvailable(self, data, stream=None):
        #self.debug("_dataAvailable: num chars %d\n", len(data))
        self.data = data
        # Add the data to the stream and notify our listener.

        if stream is None:
            # Use the default content stream.
            stream = self.contentStream
        stream.write(data)
        self._listener.onDataAvailable(None,
                                       None,
                                       stream,
                                       0,
                                       stream.available())

    def _stopRequest(self, status=None):
        self.debug("_stopRequest:")
        if status is not None:
            self.status = status
        if not self._sentStartRequest:
            # We must have sent the start request before a stop request.
            self._startRequest()
        self._listener.onStopRequest(None, self._context, self.status)

    security_level = components.interfaces.nsIAuthPrompt2.LEVEL_SECURE

    def _updateUsernameAndPassword(self, nsUri, username, password, num_tries):
        result = True

        if num_tries == 1 and username:
            # On the first time, we don't need to prompt if we have a username.
            pass

        elif not username or not password:
            # Prompt for username and password.
            authinfo = sftpAuthInfo(username=username, password=password,
                                    realm=nsUri.prePath)
            if num_tries == 1 and username:
                # Just prompt for the password this time.
                authinfo.flags |= components.interfaces.nsIAuthInformation.ONLY_PASSWORD
            result = not self._listener.promptAuth(self.security_level, authinfo)
            username = authinfo.username
            password = authinfo.password
            
        return result, username, password

    def _auth_using_agent(self, transport, username):
        """Try authorizing using the loaded public/private keys.
        
        Returns True when authentication is successful, False otherwise.
        """
        agent = paramiko.Agent()
        agent_keys = agent.get_keys()
        if len(agent_keys) == 0:
            self.debug('Login using agent was unsuccessful, no keys available.')
            return False
            
        for key in agent_keys:
            #from binascii import hexlify
            #self.debug('Trying ssh-agent key %s', hexlify(key.get_fingerprint()))
            try:
                transport.auth_publickey(username, key)
                self.debug('Login using agent key was successful.')
                return True
            except paramiko.SSHException:
                # This key didn't work... try the next one.
                pass
        self.debug('Login using agent keys was unsuccessful.')
        return False

    ##################################
    #   sftp controls
    ##################################

    def _connect_transport(self, nsUri):
        self.debug("_connect")
        self._state = self.STATE_OPENING
        username = nsUri.username
        password = nsUri.password
        host = nsUri.host
        port = nsUri.port
        if port == -1:
            port = 22

        # Try to get an already opened connection from the cache.
        transport = connectionCache.getConnection(host, port, username)
        if transport:
            try:
                self.status = nsError.NS_OK
                return transport
            except socket.error, ex:
                self.debug("Could not use the cached connection '%s:%s': %r",
                           host, username, ex)

        # Keep prompting until the user decides to cancel or major error occurs.
        num_tries = 0
        while num_tries < 20:  # Well, I think you should give up now.
            num_tries += 1
            try:

                # Connect and verify our existance.
                # TODO: This could be optimized, I dont think there is need to
                #       create a whole new socket connection... could re-use it?
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                # TODO: Pref this timeout value? Get from mozilla?
                sock.settimeout(30)
                sock.connect((host, port))

                transport = paramiko.Transport(sock)
                try:
                    transport.start_client()
                except paramiko.SSHException:
                    self.debug('Stopping request: SSH negotiation failed for start_client.')
                    self.status = nsError.NS_ERROR_FAILURE
                    return None

                # Ensure we have a username and password.
                result, username, password = self._updateUsernameAndPassword(nsUri, username, password, num_tries)
                if not result:
                    self.debug("Stopping request: user cancelled username/password prompt")
                    self.status = nsError.NS_ERROR_FAILURE
                    return None
                if not username:
                    self.debug("Stopping request: no username provided")
                    self.status = nsError.NS_ERROR_FAILURE
                    return None

                # Login, try using public/private keys the first time around.
                if num_tries > 1 or not self._auth_using_agent(transport, username):
                    # This will throw an AuthenticationException if it fails.
                    transport.auth_password(username, password)

                # Login was successful.

                # IMPORTANT!!!
                # TODO: IMPORTANT: Check server's host key in order to stop
                #       stop man-in-the-middle attacks.
                #key = transport.get_remote_server_key()

                connectionCache.addConnection(transport, host, port, username)
                self.status = nsError.NS_OK
                return transport

            except paramiko.AuthenticationException, ex:
                # Raise NS_ERROR_FTP_LOGIN
                self.debug(ex)
                self.status = nsError.NS_ERROR_GENERATE_FAILURE(nsError.NS_ERROR_MODULE_NETWORK, 21)
                # Try again...
            except socket.error, ex:
                self.debug(ex)
                self.status = nsError.NS_ERROR_GENERATE_FAILURE(nsError.NS_ERROR_MODULE_NETWORK, 21)
                return None
        return None

    def _close_sftp(self):
        if self._sftp_file is not None:
            self._sftp_file.close()
        self._sftp_file = None
        self._state = self.STATE_CLOSED

    def _upload(self, nsUri):
        self.debug("uploading")

        self._startRequest()
        # Open the file for writing
        filename = nsUri.path
        self._sftp_file = self._sftp.file(filename, "wb")
        inputstream = self.uploadStream
        while self.isPending():
            data = inputstream.read(self.READ_BLOCK_SIZE)
            if not data:
                break

            # Suspend/resume handling.
            if self._state == self.STATE_SUSPENDED:
                self._state_notification.acquire()
                try:
                    if self._state == self.STATE_SUSPENDED:
                        self._state_notification.wait()
                finally:
                    self._state_notification.release()

            self._sftp_file.write(data)
        inputstream.close()
        self._sftp_file.close()
        # Clear the file handle, so another upload/download can be performed.
        self._sftp_file = None

    def _downloadDirectoryContents(self, nsUri):
        self.debug("_downloadDirectoryContents")
        # The format for the return fields is described here:
        # http://www.mozilla.org/projects/netlib/dirindexformat.html

        data_blocks = []
        # First, take out the password field so it's not visible.
        tempUri = nsUri.clone()
        tempUri.userPass = tempUri.username;
        data_blocks.append("300: %s\r\n" % (tempUri.spec, ))
        data_blocks.append("200: filename content-length last-modified file-type\r\n")
        data_blocks.append("301: UTF-8\r\n")
        # listdir_attr() returns unsorted names, so we need to sort them.
        entries = self._sftp.listdir_attr(nsUri.path)
        import operator
        entries.sort(key=operator.attrgetter("filename"))

        for child_stat in entries:
            file_type = "FILE"
            st_mode = child_stat.st_mode
            if stat.S_ISDIR(st_mode):
                file_type = "DIRECTORY"
            elif stat.S_ISLNK(st_mode):
                file_type = "SYMBOLIC-LINK"
            date_modified = strftime("%a, %d %b %Y %H:%M:%S GMT",
                                        gmtime(child_stat.st_mtime))
            data_blocks.append("201: %s %d %s %s\n" % (
                                        quote(child_stat.filename.encode("utf-8")),
                                        child_stat.st_size,
                                        quote(date_modified),
                                        file_type))
        return "".join(data_blocks)

    def _download(self, max_read=None):
        self.debug("_download")
        self._state = self.STATE_PROCESSING

        nsUri = self.URI
        filepath = quote(nsUri.path)
        file_stat = self._sftp.stat(filepath)

        if stat.S_ISDIR(file_stat.st_mode):
            # We are just fetching the directory listing.
            self.isDirectoryListing = True
            #self.contentCharset = "ISO-8859-1"
            #self.contentCharset = "UTF-8"
            self._startRequest(self.APPLICATION_HTTP_INDEX_FORMAT)
            data = self._downloadDirectoryContents(nsUri)
            #print "data:\n%s" % (data, )
            self._dataAvailable(data)
        else:
            # We are just fetching a file's content, open for reading.
            self.fileSize = file_stat.st_size
            sftp_file = self._sftp.file(filepath)

            # Resume handling.
            entityID = "%s:%s" % (self.fileSize, file_stat.st_mtime)
            if self.entityID:
                # It's a resume operation.
                self.debug("resume:: entityID: %r", self.entityID)
                if self.entityID != entityID:
                    # The file must have changed.
                    self.debug("resume:: file stats have changed, cannot resume")
                    raise ServerException(self.NS_ERROR_NOT_RESUMABLE)
                if self._startPos > 0:
                    try:
                        self.debug("resume:: resuming at: %r", self._startPos)
                        sftp_file.seek(self._startPos)
                    except IOError:
                        self.debug("resume:: file does not support resume")
                        raise ServerException(self.NS_ERROR_NOT_RESUMABLE)

            self.entityID = entityID
            try:
                # Start reading as much data as possible asynchronously.
                sftp_file.prefetch()
                self._startRequest(self.APPLICATION_GUESS_FROM_EXT)
                while 1:
                    data = sftp_file.read(self.READ_BLOCK_SIZE)
                    if not data:
                        break

                    # Suspend/resume handling.
                    if self._state == self.STATE_SUSPENDED:
                        self._state_notification.acquire()
                        try:
                            if self._state == self.STATE_SUSPENDED:
                                self._state_notification.wait()
                        finally:
                            self._state_notification.release()

                    if self._state != self.STATE_PROCESSING:
                        break
                    # Send the data to the listener.
                    self._dataAvailable(data)
            finally:
                # This is a special Paramiko close file method, it's used to
                # ensure we don't have to wait for the prefetch call to finish
                # before the file handle is closed.
                #sftp_file.close()
                sftp_file._close(async=True)

    def _run_command(self, channel, command):
        self.debug("_run_command: %r", command)
        try:
            channel.exec_command(command)
        except paramiko.SSHException:
            self.debug('Stopping request: SSH exec_command failed.')
            self.status = nsError.NS_ERROR_FAILURE
            return

        self._state = self.STATE_PROCESSING
        self._startRequest(actionType="run_command")

        # Setup stderr reader thread.
        def _read_stderr():
            stderrStream = SSHStderrStream()
            while True:
                data = channel.recv_stderr(self.READ_BLOCK_SIZE)
                if not data:
                    break
                if self._state != self.STATE_PROCESSING:
                    break
                # Send the data to the listener.
                self._dataAvailable(data, stream=stderrStream)
        stderr_thread = threading.Thread(target=_read_stderr)
        stderr_thread.setDaemon(True)
        stderr_thread.start()

        # Read the stdout data.
        while True:
            data = channel.recv(self.READ_BLOCK_SIZE)
            if not data:
                break
            if self._state != self.STATE_PROCESSING:
                break
            # Send the data to the listener.
            self._dataAvailable(data)
        # Wait till stderr is done as well.
        stderr_thread.join()

    def _interact(self, channel):
        self.debug("_interact")
        # Hook the channel IO.
        try:
            channel.get_pty(self._termType)
        except paramiko.SSHException:
            self.debug('Stopping request: SSH negotiation failed for invoke_pty.')
            self.status = nsError.NS_ERROR_FAILURE
            return
        try:
            channel.invoke_shell()
        except paramiko.SSHException:
            self.debug('Stopping request: SSH negotiation failed for invoke_shell.')
            self.status = nsError.NS_ERROR_FAILURE
            return

        self._state = self.STATE_PROCESSING
        self._startRequest()
        # Read the data.
        while True:
            data = channel.recv(self.READ_BLOCK_SIZE)
            if not data:
                break
            if self._state != self.STATE_PROCESSING:
                break
            # Send the data to the listener.
            self._dataAvailable(data)

    def _cleanup(self):
        # Null out some variables... call me paranoid :)
        self._sftp = None
        self._sftp_file = None
        self._channel = None
        self._targetThread = None
        self._proxiedAuthSvc = None
        self._context = None
        self._listener = None
        self.contentStream = None
        self.uploadStream = None

    ##################################
    #   python thread method
    ##################################

    def run(self):
        self.debug("run")

        try:
            nsUri = self.URI
            transport = self._connect_transport(nsUri)
            if self.status != nsError.NS_OK or transport is None:
                return
            if self._async_type == self.TYPE_ASYNC_OPEN:
                self._sftp = paramiko.SFTPClient.from_transport(transport)
                try:
                    if self.uploadStream:
                        self._upload(nsUri)
                    else:
                        self._download(nsUri)
                finally:
                    self._close_sftp()
                    self.debug("sftp handle is closed")

            elif self._async_type == self.TYPE_ASYNC_COMMAND:
                self._channel = transport.open_session()
                try:
                    self._run_command(self._channel, self._command)
                finally:
                    self._channel.close()

            elif self._async_type == self.TYPE_ASYNC_INTERACT:
                self._channel = transport.open_session()
                try:
                    self._interact(self._channel)
                finally:
                    self._channel.close()

        except Exception, ex:
            self.debug("SFTPProtocol.run() exception...")
            import traceback
            traceback.print_exc()
            self.status = nsError.NS_ERROR_FAILURE

        finally:
            self._state = self.STATE_CLOSED
            self._stopRequest()
            self._cleanup()
            self.debug("Done")


    ##################################
    #   pythonextISSHRunCommand and pyISSHInteraction
    ##################################

    def asyncRunCommand(self, aListener, aCommand, aContext):
        self.debug("asyncRunCommand")
        if self._state != self.STATE_INITIALIZED:
            raise ServerException(nsError.NS_ERROR_ALREADY_OPENED)

        self._async_type = self.TYPE_ASYNC_COMMAND
        self._listener = getProxyForObject(self._targetThread,
                                           components.interfaces.pyISSHStreamListener,
                                           aListener,
                                           NS_PROXY_SYNC | NS_PROXY_ALWAYS)
        self._command = aCommand
        self._context = aContext
        self._state = self.STATE_OPENING

        threadMgr = components.classes["@mozilla.org/thread-manager;1"].\
                        getService(components.interfaces.nsIThreadManager)
        self._targetThread = threadMgr.currentThread

        # Kick off the asynchronous Python thread (this calls the run() method).
        self.start()

    def asyncInteract(self, aListener, aTerminalType, aContext):
        self.debug("asyncInteraction")
        if self._state != self.STATE_INITIALIZED:
            raise ServerException(nsError.NS_ERROR_ALREADY_OPENED)

        self._async_type = self.TYPE_ASYNC_INTERACT
        self._listener = getProxyForObject(self._targetThread,
                                           components.interfaces.pyISSHStreamListener,
                                           aListener,
                                           NS_PROXY_SYNC | NS_PROXY_ALWAYS)
        self._context = aContext
        self._termType = aTerminalType
        self._state = self.STATE_OPENING

        threadMgr = components.classes["@mozilla.org/thread-manager;1"].\
                        getService(components.interfaces.nsIThreadManager)
        self._targetThread = threadMgr.currentThread

        # Kick off the asynchronous Python thread (this calls the run() method).
        self.start()

    def sendStdin(self, aData):
        if self._state != self.STATE_PROCESSING or not self._channel:
            self.debug("sendStdin:: invalid connection state: %d", self._state)
            raise ServerException(nsError.NS_ERROR_FAILURE)
        self._channel.sendall(aData)
